package process;


import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import common.Const;
import configer.SqlServerConfiger;
import configer.SqlServerInitConfiger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import subs.SqlServerSub;
import sync.SqlServerSync;
import utils.Druid;
import utils.OperationTxt;
import utils.Tools;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class Prepare {

    protected final static Logger logger = LoggerFactory.getLogger(Prepare.class);     // 日志对象

    public Map<String, String> batchServerFlag = new HashMap<>();
    private Map<String, String> batchSqlServerLsn = new HashMap<>();
    private Map<String, String> batchPage = new HashMap<>();


    public void run() {
        // 初始化参数
        SqlServerInitConfiger sqlServerInitConfiger = new SqlServerInitConfiger(Const.PROPERTIES_NAME);
        List<SqlServerConfiger> configers = sqlServerInitConfiger.getConfigrs();

        OperationTxt sqlServerFlag = new OperationTxt(Const.SQLSERVER_FLAG);
        OperationTxt sqlServerLsn = new OperationTxt(Const.SQLSERVER_LSN);
        OperationTxt sqlServerPage = new OperationTxt(Const.SQLSERVER_PAGE);

        // 初始化Map（从txt中读取）
        if (!"{}".equals(sqlServerLsn.getContent())) {
            Map newMap = JSONUtil.parseObj(sqlServerLsn.getContent());
            batchSqlServerLsn.putAll(newMap);
        }
        if (!"{}".equals(sqlServerPage.getContent())) {
            Map newMap = JSONUtil.parseObj(sqlServerPage.getContent());
            batchPage.putAll(newMap);
        }
        if (!"{}".equals(sqlServerFlag.getContent())) {
            Map newMap = JSONUtil.parseObj(sqlServerFlag.getContent());
            batchServerFlag.putAll(newMap);
        }

        // 持久化
        mapIntoTxtThread(sqlServerLsn, batchSqlServerLsn);
        mapIntoTxtThread(sqlServerPage, batchPage);

        for (SqlServerConfiger sqlServerConfiger : configers) {
            //开启线程
            Runnable r = () -> {

                // 验证传入的参数是否合理
                argValidate(sqlServerConfiger);

                Boolean flag = batchServerFlag.containsKey(sqlServerConfiger.getSqlServer_database()
                        + "." + sqlServerConfiger.getSqlServer_schema()
                        + "." + sqlServerConfiger.getSqlServer_tableName());
                if (!flag) {
                    Boolean isFinishSqlServerSync = sqlServerSync(sqlServerConfiger);
                    synchronized (batchServerFlag) {
                        batchServerFlag.put(sqlServerConfiger.getSqlServer_database()
                                + "." + sqlServerConfiger.getSqlServer_schema()
                                + "." + sqlServerConfiger.getSqlServer_tableName(), "true");
                        sqlServerFlag.setContent(JSONUtil.toJsonStr(batchServerFlag));
                    }
                    if (isFinishSqlServerSync) {
                        sqlServerSub(sqlServerConfiger);
                    }
                } else {
                    sqlServerSub(sqlServerConfiger);
                }
            };
            ThreadUtil.execute(r);
        }
    }


    /**
     * @Descriotion 每隔5分钟将Map中的数据持久化到txt
     */
    public void mapIntoTxtThread(OperationTxt operationTxt, Map<String, String> batchmap) {
        Runnable r = () -> {
            while (true) {
                synchronized (batchmap) {
                    operationTxt.setContent(JSONUtil.toJsonStr(batchmap));
                    logger.info("将Map中的数据持久化到txt,内容为:" + JSONUtil.toJsonStr(batchmap));
                }
                try {
                    //休眠5分钟
                    Thread.sleep(300000);
                } catch (InterruptedException e) {
                    logger.error("Thread.sleep 失败 { " + e + " } ");
                }
            }
        };
        ThreadUtil.execute(r);
    }


    /**
     * @Descriotion SqlServer数据库全量同步
     */
    public boolean sqlServerSync(SqlServerConfiger sqlServerConfiger) {
        // 实例化 <SqlServer数据库全量同步> 处理对象
        SqlServerSync sqlServerSync = new SqlServerSync(sqlServerConfiger, batchPage);
        // 创建 数据库连接池
        Druid druid = createDruid(sqlServerConfiger);
        // 预测试程序能否正常运行
        sqlServerSync.testRun(druid);
        // 开始监听，进行全量同步
        sqlServerSync.run(druid);
        return true;
    }


    /**
     * @Descriotion SqlServer数据库增量订阅
     */
    public void sqlServerSub(SqlServerConfiger sqlServerConfiger) {
        // 初始化SqlServer参数
        OperationTxt sqlServerLsn = new OperationTxt(Const.SQLSERVER_LSN);
        // 实例化 <SqlServer数据库增量订阅> 处理对象
        SqlServerSub sqlServerSub = new SqlServerSub(sqlServerConfiger, batchSqlServerLsn);
        // 创建 数据库连接池
        Druid druid1 = createDruid(sqlServerConfiger);
        // 预测试程序能否正常运行
        sqlServerSub.testRun(druid1);
        // 进行正式的 数据库增量订阅
        sqlServerSub.run(druid1);
    }


    public Druid createDruid(SqlServerConfiger sqlServerConfiger) {

        Properties properties = new Properties();           // Druid创建时使用的properties对象

        // 设置Druid需要的参数
        properties.setProperty("driver", Const.SQLSERVER_DRIVER_NAME);
        properties.setProperty("username", sqlServerConfiger.getSqlServer_account());
        properties.setProperty("password", sqlServerConfiger.getSqlServer_password());
        properties.setProperty("url", "jdbc:sqlserver://" + sqlServerConfiger.getSqlServer_address() + ":" + sqlServerConfiger.getSqlServer_port());

        // 创建Druid对象
        Druid druid = new Druid(properties);
        return druid;
    }


    /**
     * @Descriotion 验证传入的参数是否合理
     */
    public void argValidate(SqlServerConfiger sqlServerConfiger) {

        logger.info("请核对 数据库 配置是否正确");
        logger.info("数据库IP地址：" + sqlServerConfiger.getSqlServer_address());
        logger.info("数据库访问端口：" + sqlServerConfiger.getSqlServer_port());
        logger.info("数据库访问账户：" + sqlServerConfiger.getSqlServer_account());
        logger.info("数据库账户密码：" + sqlServerConfiger.getSqlServer_password());
        logger.info("数据库名称：" + sqlServerConfiger.getSqlServer_database());
        logger.info("数据库模式：" + sqlServerConfiger.getSqlServer_schema());
        logger.info("数据表名称：" + sqlServerConfiger.getSqlServer_tableName());

        logger.info("请核对 head 配置是否正确");
        logger.info("元数据名称：" + sqlServerConfiger.getDataName());
        logger.info("数据类型：" + sqlServerConfiger.getDataType());
        logger.info("数据索引表：" + sqlServerConfiger.getDataIndexs());
        logger.info("上链模式：" + sqlServerConfiger.getDataFlag());
        logger.info("加密方式：" + sqlServerConfiger.getDataCrypt());
        logger.info("数据级别：" + sqlServerConfiger.getDataLevel());

        logger.info("请核对 rocketmq 配置是否正确");
        logger.info("NameSrv地址：" + sqlServerConfiger.getNameSrv());
        logger.info("Topic名称：" + sqlServerConfiger.getTopicName());
        logger.info("Tag名称：" + sqlServerConfiger.getTagName());
        logger.info("生产者组：" + sqlServerConfiger.getProducerGroup());

        Tools.getUserYorN();

    }

}
